package com.hksj.Intelligentmanagement.mqtt.callback;

import com.hksj.Intelligentmanagement.mqtt.client.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MqttCallback implements MqttCallbackExtended {

    private static final Logger log = LoggerFactory.getLogger(MqttCallback.class);

    private IMqttClient mqttClient;

    private MqttCallBackAdapter callBackAdapter;

    public MqttCallback(IMqttClient mqttClient, MqttCallBackAdapter callBackAdapter) {
        this.mqttClient = mqttClient;
        this.callBackAdapter = callBackAdapter;
    }


    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connectionLost 连接断开，每隔5S进行尝试重连: {}", throwable.getMessage());
        callBackAdapter.getExecutor().execute(() -> {
            long reconnectTimes = 1;
            while (!mqttClient.getClient().isConnected()) {
                try {
                    reconnectTimes += 1;
                    log.warn("mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
                    mqttClient.getClient().reconnect();
                } catch (MqttException e) {
                    //重连失败后休眠5秒
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException e1) {
                    }
                }
            }
        });
    }

    /**
     * @param topic
     * @param mqttMessage
     * @throws Exception
     * subscribe后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
//        log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
        //发布消息主题
        if (callBackAdapter != null){
            callBackAdapter.handleMsg(topic,mqttMessage);
        }
    }

    /**
     * 连接成功后的回调 可以在这个方法执行 订阅主题  生成Bean的 MqttConfiguration方法中订阅主题 出现bug
     * 重新连接后  主题也需要再次订阅  将重新订阅主题放在连接成功后的回调 比较合理
     * @param reconnect
     * @param serverURI
     */
    @Override
    public void connectComplete(boolean reconnect,String serverURI){
        log.info("MQTT {} 连接成功，连接方式：{}",serverURI,reconnect ? "重连" : "直连");
        if (callBackAdapter != null){
            callBackAdapter.initSubscript(mqttClient);
        }
    }

    /**
     * 消息到达后
     * publish后，配送完成后回调的方法
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }
}
